package com.yzy.fly.controller;

import com.yzy.fly.api.annotation.ApiDoc;
import com.yzy.fly.api.annotation.ApiOperation;
import com.yzy.fly.api.function.response.entity.FlyApiResponseEntity;
import com.yzy.fly.api.function.response.entity.FlyBasicApiResponseEntity;
import com.yzy.fly.exception.ApiResponseException;
import com.yzy.fly.kernel.controller.FlyConstant;
import com.yzy.fly.kernel.controller.FlyNoneController;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * kafka生产者【实际上就是一个Controller，用来进行消息生产】
 */
@ApiDoc(desc = "kafka消息")
@RestController
public class KafkaProducerController extends FlyNoneController {
    private final static String TOPIC_NAME = "zhTest"; //topic的名称

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

//    @ApiOperation(value = "JSON合法性",notes = "判断内容是否为JSON的正确格式",author="yinzhiyuan")
//    @RequestMapping("/send")
//    public void send() {
//        kafkaTemplate.send(TOPIC_NAME,  "key", "test message send");
//    }

    @ApiOperation(value = "消息生产",notes = "生产kafka消息",author="yinzhiyuan")
    @RequestMapping("/send")
    public FlyApiResponseEntity send(@RequestParam String topic, @RequestParam String key, @RequestParam String data) {
        long statTime = System.currentTimeMillis();
        if (StringUtils.isBlank(topic)) {
            throw new ApiResponseException(500,"主题不能定义为空");
        }
        if (StringUtils.isBlank(key)) {
            throw new ApiResponseException(500,"主题不能定义为空");
        }
        if (StringUtils.isBlank(data)) {
            throw new ApiResponseException(500,"主题不能定义为空");
        }
        kafkaTemplate.send(TOPIC_NAME,  "key", "test message send");
        long endTime = System.currentTimeMillis();
        return FlyBasicApiResponseEntity.success("发送成功",null,(endTime-statTime));
    }

    @Override
    public String getType() {
        return FlyConstant.FLY_CORE;
    }
}
